package march.firstmarch21.march01.plan2;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;
import march.firstmarch21.march01.CallsBean;

import java.io.IOException;
import java.text.SimpleDateFormat;
import java.util.HashMap;

public class CallsOtherMethod extends Configured implements Tool {
    public static HashMap<String, String> location = new HashMap<>();
    public static HashMap<String, String> userPhone = new HashMap<>();

    static class CallsMapper02 extends Mapper<LongWritable, Text, NullWritable,CallsBean> {
        private Text key=new Text();

        String fileName;
        //该方法会在map执行前，执行一次
        @Override
        protected void setup(Context context) throws IOException, InterruptedException {
            //获取文件名
            FileSplit inputSplit = (FileSplit) context.getInputSplit();
            fileName = inputSplit.getPath().getName();

        }

        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {

            //1.每行数据拿到并做上标志
            String[] strings = value.toString().split(",");

            //分类处理
            if(fileName.contains("calls.txt")){
                CallsBean callsBean = new CallsBean();
                //2.解决时间格式和通话时长
                String start = strings[2];
                String end = strings[3];
                //获取时间差
                String time = Long.parseLong(end)-Long.parseLong(start)+"秒";
                //修改时间格式
                SimpleDateFormat sdf = new SimpleDateFormat("yyyy年MM月dd日 HH:mm:ss");
                start = sdf.format(Long.parseLong(start));
                end = sdf.format(Long.parseLong(end));
                //对象填充
                callsBean.setCallName(strings[0]);
                callsBean.setAcceptName(strings[1]);
                callsBean.setStartTime(start);
                callsBean.setEndTime(end);
                callsBean.setCallLocation(strings[4]);
                callsBean.setAcceptLocation(strings[5]);
                callsBean.setTime(time);

                context.write(NullWritable.get(),callsBean);
            }else if(fileName.contains("location.txt")){
                location.put(strings[1],strings[2]);
            }else if(fileName.contains("userPhone.txt")){
                userPhone.put(strings[1],strings[2]);
            }
        }
    }
    static class CallsReducer02 extends Reducer<NullWritable, CallsBean,NullWritable,CallsBean> {

        @Override
        protected void reduce(NullWritable key, Iterable<CallsBean> values, Context context) throws IOException, InterruptedException {

            for (CallsBean value : values) {


                String callName = userPhone.get(value.getCallName());
                String acceptName = userPhone.get(value.getAcceptName());
                String callLocation = location.get(value.getCallLocation());
                String acceptLocation = location.get(value.getAcceptLocation());

                value.setCallName(callName);
                value.setAcceptName(acceptName);
                value.setCallLocation(callLocation);
                value.setAcceptLocation(acceptLocation);

                context.write(NullWritable.get(),value);
            }

        }
    }
    @Override
    public int run(String[] args) throws Exception {
        Configuration conf=getConf();
//        Path input=new Path(conf.get("input"));
//        Path output=new Path(conf.get("output"));

        Job job=Job.getInstance(conf,this.getClass().getSimpleName());
        job.setJarByClass(CallsOtherMethod.class);

        job.setMapperClass(CallsMapper02.class);
        job.setMapOutputKeyClass(NullWritable.class);
        job.setMapOutputValueClass(CallsBean.class);

        job.setReducerClass(CallsReducer02.class);
        job.setOutputKeyClass(NullWritable.class);
        job.setOutputValueClass(CallsBean.class);

        job.setInputFormatClass(TextInputFormat.class);
        job.setOutputFormatClass(TextOutputFormat.class);

        TextInputFormat.addInputPath(job,new Path("F:\\test\\test4Input"));
        TextOutputFormat.setOutputPath(job,new Path("F:\\test\\test5Output"));

        return job.waitForCompletion(true)?0:1;
    }

    public static void main(String[] args) throws Exception {

//        args = new String[]{"F:\\test\\input", "F:\\test\\output"};
        System.exit(ToolRunner.run(new CallsOtherMethod(),args));
    }

}
